package com.atguigu.day04;

import com.atguigu.utils.IntegerSource;
import com.atguigu.utils.IntegerStatistic;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntegerSource())
                .keyBy(r -> "int")
                .process(new KeyedProcessFunction<String, Integer, IntegerStatistic>() {
                    private ValueState<IntegerStatistic> accumulator;
                    // 标志位状态变量，标志位如果不为空，说明存在定时器
                    private ValueState<Integer> flag;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        accumulator = getRuntimeContext().getState(
                                new ValueStateDescriptor<IntegerStatistic>(
                                        "accumulator",
                                        Types.POJO(IntegerStatistic.class)
                                )
                        );
                        flag = getRuntimeContext().getState(
                                new ValueStateDescriptor<Integer>(
                                        "flag",
                                        Types.INT
                                )
                        );
                    }

                    @Override
                    public void processElement(Integer in, Context ctx, Collector<IntegerStatistic> out) throws Exception {
                        if (accumulator.value() == null) {
                            accumulator.update(new IntegerStatistic(in, in, in, 1, in));
                        } else {
                            IntegerStatistic oldAccumulator = accumulator.value();
                            IntegerStatistic newAccumulator = new IntegerStatistic(
                                    Math.min(oldAccumulator.min, in),
                                    Math.max(oldAccumulator.max, in),
                                    oldAccumulator.sum + in,
                                    oldAccumulator.count + 1,
                                    (oldAccumulator.sum + in) / (oldAccumulator.count + 1)
                            );
                            accumulator.update(newAccumulator);
                        }

                        // 检查定时器是否存在
                        // 如果定时器不存在
                        // 第一步：注册定时器
                        // 第二步：标志位置为非空
                        if (flag.value() == null) {
                            ctx.timerService().registerProcessingTimeTimer(
                                    ctx.timerService().currentProcessingTime() + 10 * 1000L
                            );
                            flag.update(1);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<IntegerStatistic> out) throws Exception {
                        out.collect(accumulator.value());
                        // 标志位置为空
                        flag.clear();
                    }
                })
                .print();

        env.execute();
    }
}
